#include "stream_handle.h"
#include <sstream>
#include <iostream>
#include "event2/bufferevent.h"
#include "event2/event.h"
#include "../security_msg.pb.h"
#include "../com/simple_logger.h"
#include "stream_service.h"
#include "../globe_type.h"

using namespace std;
using namespace sec;

static const string STORAGE = "temp/";
// 读回调
void SReadCb(struct bufferevent *buffev, void *ctx)
{
    StreamHandle *handle = (StreamHandle *)ctx;
    handle->ReadCb();
}

// 写回调
static void SWriteCb(struct bufferevent *buffev, void *ctx)
{
    StreamHandle *handle = (StreamHandle *)ctx;
    handle->WriteCb();
}

// 事件回调
static void SEventCb(struct bufferevent *buffev, short what, void *ctx)
{
    StreamHandle *handle = (StreamHandle *)ctx;
    handle->EventCb(what);
}

StreamHandle::StreamHandle(int socket, StreamService *parent)
    : socket_(socket), parent_(parent), is_stream_(false), remain_size_(0)
{
}

StreamHandle::~StreamHandle()
{
}

void StreamHandle::ReadCb()
{
    if (!is_stream_)
    {
        ProtoBufRead();
    }
    else
    {
        ByteStreamRead();
    }
}

void StreamHandle::WriteCb()
{
}

void StreamHandle::EventCb(short which)
{
    if (which & BEV_EVENT_EOF || which & BEV_EVENT_ERROR)
    {
        if (bev_)
        {
            bufferevent_free(bev_);
            bev_ = nullptr;
        }
    }
    if (which & BEV_EVENT_TIMEOUT)
    {
        if (is_stream_)
        {
            LOG4CPLUS_DEBUG(SimpleLogger::Get()->LoggerRef(), "流状态下 超时事件将重新注册读写操作");
            bufferevent_enable(bev_, EV_READ | EV_WRITE);
        }
        else
        {
            bufferevent_free(bev_);
            bev_ = nullptr;
        }
    }
}

bool StreamHandle::Response(const char *header, int header_size, const char *data, int data_size)
{
    int ret = bufferevent_write(bev_, PKT_FLAG, 4);
    if (ret == -1)
    {
        return false;
    }
    ret = bufferevent_write(bev_, &header_size, 4);
    if (ret == -1)
    {
        return false;
    }
    ret = bufferevent_write(bev_, header, header_size);
    if (ret == -1)
    {
        return false;
    }
    if (data_size > 0)
    {
        ret = bufferevent_write(bev_, data, data_size);
        if (ret == -1)
        {
            return false;
        }
    }
    return true;
}

void StreamHandle::SendFile()
{
    if (!pf_)
    {
        stringstream ss;
        ss << STORAGE << this->file_name_;
        this->pf_ = fopen(ss.str().c_str(), "rb");
    }
    if (!pf_)
    {
        stringstream ss;
        ss << STORAGE << this->file_name_ << " 文件无法打开";
        LOG4CPLUS_ERROR(SimpleLogger::Get()->LoggerRef(), ss.str().c_str());
        return;
    }
    fseek(pf_, 0, SEEK_END);
    this->file_size_ = ftell(pf_);
    fseek(pf_, 0, SEEK_SET);
    static int send_size = 0;
    char buf[1024] = {0};
    while (1)
    {
        int len = fread(buf, 1, sizeof(buf), pf_);
        if (len > 0)
        {
            bufferevent_write(bev_, buf, len);
            send_size += len;
        }
        if (len == 0)
        {
            is_stream_ = false;
            send_size = 0;
            fclose(pf_);
            pf_ = nullptr;
            break;
        }
    }
}

void StreamHandle::Launch()
{
    if (eb_ || bev_)
    {
        return;
    }
    eb_ = event_base_new();
    bev_ = bufferevent_socket_new(eb_, socket_, BEV_OPT_CLOSE_ON_FREE);
    if (!bev_)
    {
        return;
    }
    // 指定参数
    bufferevent_setcb(bev_, SReadCb, SWriteCb, SEventCb, this);
    bufferevent_enable(bev_, EV_READ | EV_WRITE);
    timeval tv = {10, 0};
    bufferevent_set_timeouts(bev_, &tv, &tv);
    event_base_dispatch(eb_);
    if (eb_)
    {
        event_base_free(eb_);
        eb_ = nullptr;
    }

    LOG4CPLUS_DEBUG(SimpleLogger::Get()->LoggerRef(), "流连接断开");
    delete this;
}

void StreamHandle::ProtoBufRead()
{
    char flag[5] = {0};
    int head_size = 0;
    // 判断报文头
    int len = bufferevent_read(bev_, flag, 4);
    if (len <= 0 || strcmp(flag, PKT_FLAG) != 0)
    {
        return;
    }
    // 获得消息头大小
    len = bufferevent_read(bev_, &head_size, 4);
    if (len <= 0)
    {
        return;
    }
    char *head = new char[head_size];
    len = bufferevent_read(bev_, head, head_size);
    // 解析消息头对象
    MsgHeader msg_header;
    msg_header.ParseFromArray(head, head_size);
    delete[] head;
    char *msg_buff = new char[msg_header.msg_size()];
    // FIX:if msg_body too large
    len = bufferevent_read(bev_, msg_buff, msg_header.msg_size());
    switch (msg_header.msg_type())
    {
    case MsgType::FILE_UPLOAD_REQ:
    {
        FileUploadReq req;
        req.ParseFromArray(msg_buff, len);
        this->file_name_ = req.file_name();
        this->file_size_ = req.file_size();
        this->signature_ = req.signature();
        this->send_id_ = msg_header.from();
        this->recv_id_ = msg_header.to();
        FileUploadResp resp;
        resp.set_resp_type(FileUploadResp_FileUploadRespType_OK);
        MsgHeader resp_header;
        resp_header.set_msg_size(resp.ByteSizeLong());
        resp_header.set_msg_type(MsgType::FILE_UPLOAD_RESP);
        resp_header.set_from(0);
        resp_header.set_to(msg_header.from());
        this->is_stream_ = true; //
        this->remain_size_ = 0;
        this->Response(resp_header.SerializeAsString().c_str(), resp_header.ByteSizeLong(), resp.SerializeAsString().c_str(), resp.ByteSizeLong());
        break;
    }
    case MsgType::FILE_DOWNLOAD_REQ:
    {
        FileDownloadReq req;
        req.ParseFromArray(msg_buff, len);
        this->file_name_ = req.file_name();
        this->is_stream_ = true;
        SendFile();
        break;
    }
    case MsgType::FILE_DOWNLOAD_RES:
    {
        FileDownloadRes res;
        res.ParseFromArray(msg_buff, len);
        if (res.res_type() == FileDownloadRes_FileDownloadResType::FileDownloadRes_FileDownloadResType_OK)
        {
            stringstream ss;
            ss << STORAGE << this->file_name_;
            remove(ss.str().c_str());
        }
        break;
    }
    default:
        stringstream ss;
        ss << "事件类型未注册：" << msg_header.msg_type();
        LOG4CPLUS_ERROR(SimpleLogger::Get()->LoggerRef(), ss.str());
    }
    delete[] msg_buff;
}

void StreamHandle::ByteStreamRead()
{
    if (!pf_)
    {
        stringstream ss;
        ss << STORAGE << this->file_name_;
        pf_ = fopen(ss.str().c_str(), "wb");
    }
    if (!pf_)
    {
        stringstream ss;
        ss << STORAGE << this->file_name_ << " 文件无法打开";
        LOG4CPLUS_ERROR(SimpleLogger::Get()->LoggerRef(), ss.str());
        return;
    }
    char buf[1024 * 10] = {0};
    while (1)
    {
        int len = bufferevent_read(bev_, buf, sizeof(buf));
        if (len > 0)
        {
            fwrite(buf, 1, len, pf_);
            remain_size_ += len;
        }
        if (len == 0)
        {
            break;
        }
    }
    if (remain_size_ == file_size_)
    {
        is_stream_ = false;
        fclose(this->pf_);
        this->pf_ = nullptr;
        //
        {
            FileUploadRes res;
            res.set_res_type(FileUploadRes_FileUploadResType::FileUploadRes_FileUploadResType_OK);
            MsgHeader res_header;
            res_header.set_from(0);
            res_header.set_to(this->send_id_);
            res_header.set_msg_size(res.ByteSizeLong());
            res_header.set_msg_type(MsgType::FILE_UPLOAD_RES);
            Response(res_header.SerializeAsString().c_str(), res_header.ByteSizeLong(), res.SerializeAsString().c_str(), res.ByteSizeLong());
        }
        {
            FileDownloadNotice notice;
            notice.set_file_name(this->file_name_.c_str());
            notice.set_file_size(this->file_size_);
            notice.set_signature(this->signature_.c_str());
            MsgHeader notice_header;
            notice_header.set_from(this->send_id_);
            notice_header.set_to(this->recv_id_);
            notice_header.set_msg_type(MsgType::FILE_DOWNLOAD_NOTICE);
            notice_header.set_msg_size(notice.ByteSizeLong());
            this->parent_->SendToCenter(notice_header.SerializeAsString().c_str(), notice_header.ByteSizeLong(), notice.SerializeAsString().c_str(), notice.ByteSizeLong());
        }
    }
}
